package com.um.springboot.starter.event;

import com.um.springboot.starter.properties.RedisEventProperties;
import lombok.extern.slf4j.Slf4j;
import org.redisson.RedissonShutdownException;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.concurrent.Executors;

/**
 * Redis分布式事件消费者
 *
 * @author 甘天伟
 * @since 2024/5/7
 * 版权所有 广州优匠科技有限公司
 */
@Slf4j
@Component
@ConditionalOnProperty("hsh.redis-event.consumer")
public class RedisEventConsumer implements ApplicationRunner {

    @Autowired
    private RedisEventProperties redisEventProperties;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private ApplicationEventPublisher applicationEventPublisher;

    private boolean destroyed = false;

    @Override
    public void run(ApplicationArguments args) {
        Executors.newSingleThreadExecutor().execute(() -> {
            log.info("启动Redis分布式事件消费者: {}", redisEventProperties.getQueueName());
            RBlockingDeque<RedisEventItem> blockingDeque = redissonClient.getBlockingDeque(redisEventProperties.getQueueName());
            while (!destroyed) {
                try {
                    RedisEventItem item = blockingDeque.take();
                    log.debug("消费事件: {}", item);

                    RedisEvent event = item.toEvent();
                    if (event != null) {
                        // 转换为本地程序事件
                        applicationEventPublisher.publishEvent(event);
                    }
                } catch (RedissonShutdownException e) {
                    log.warn("消费事件错误: {}", e.getMessage());
                    break;
                } catch (Exception e) {
                    if (!destroyed) {
                        log.error("消费事件错误", e);
                    }
                }
            }
        });
    }

    @PreDestroy
    public void destroy() {
        destroyed = true;
        log.info("停止Redis分布式事件消费者: {}", redisEventProperties.getQueueName());
    }
}
